package Kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Properties;
import java.util.Random;

public class SupermarketProducer {

    public static void main(String[] args) {
        // 设置 Kafka 生产者的配置
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.88.131:9092,192.168.88.132:9092,192.168.88.133:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 创建 Kafka 生产者
        Producer<String, String> producer = new KafkaProducer<>(props);

        // 设置日期格式
        SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");

        // 定义商品列表
        String[] products = {
                "Apple", "Banana", "Orange", "Grapes", "Mango",
                "Strawberry", "Pineapple", "Watermelon", "Cherry", "Kiwi",
                "Lemon", "Avocado", "Peach", "Pear", "Plum",
                "Blueberry", "Raspberry", "Blackberry", "Apricot", "Fig"
        };

        // 生成随机日期
        Random random = new Random();
        int minYear = 2000; // 最小年份
        int maxYear = 2023; // 最大年份
        int minMonth = 1; // 最小月份
        int maxMonth = 12; // 最大月份
        int minDay = 1; // 最小日期
        int maxDay = 28; // 最大日期（为了简化，我们假设每个月都有28天）

        try {
            while (true) { // 无限循环生成数据
                for (String product : products) { // 遍历每个商品
                    int year = minYear + random.nextInt(maxYear - minYear + 1);
                    int month = minMonth + random.nextInt(maxMonth - minMonth + 1);
                    int day = minDay + random.nextInt(maxDay - minDay + 1);

                    // 创建日期对象
                    Date date = new Date(year - 1900, month - 1, day); // 注意：Date构造函数中的年份是从1900年开始的

                    // 格式化日期为字符串
                    String productCode = sdf.format(date);

                    // 随机生成已进货数量和已售卖数量
                    int stockQuantity = random.nextInt(100) + 1; // 1 到 100 之间的随机数
                    int soldQuantity = random.nextInt(stockQuantity + 1); // 已售数量不能超过已进货数量

                    // 随机生成上架状态
                    String status = random.nextBoolean() ? "U" : "D"; // U 表示上架，D 表示下架

                    // 拼接成一条完整的消息
                    String message = productCode + "\t" + product + "\t" + stockQuantity + "\t" + soldQuantity + "\t" + status;

                    // 发送消息到 Kafka 主题
                    ProducerRecord<String, String> record = new ProducerRecord<>("supermarket", productCode, message);
                    producer.send(record);

                    System.out.println("Sent message: " + message);

                    // 模拟延迟，避免过快生成数据
                    Thread.sleep(1000); // 每秒发送一次
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            // 关闭生产者
            producer.close();
        }
    }
}